package com.orderplus.opshop.synces.listener;

import cn.hutool.core.exceptions.ExceptionUtil;
import com.alibaba.otter.canal.client.CanalConnector;
import com.orderplus.opshop.synces.core.CanalClientFactory;
import com.orderplus.opshop.synces.service.product.ProductService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;

import java.util.Map;

/**
 * Title: Rocket 自动化脚本初始化
 * Description: 自定义启动监听类
 * Company: Copyright @ 2019 OrderPlus 版权所有
 *
 * @version 1.0 初稿
 * @author: 麦豪俊
 * @date: 2019-04-02 08:09:10
 */
@Component
@Order(999)
@Slf4j
public class CanalListener extends CanalClientFactory implements CommandLineRunner, DisposableBean {
	@Autowired
	private ProductService productService;

	private CanalConnector connector = null;

	private Thread thread = null;

	//	启动监听
	@Override
	public void run(String... args) {
		runApplication();
	}

	//	关闭的时候,自动关闭连接
	@Override
	public void destroy() {
		thread.interrupt();
		connector.disconnect();
		log.info("************************** shutdown CanalClient **************************");
	}

	public void runApplication() {
		connector = super.createConnectors();

		thread = new ReaderThread();

		thread.start();

		log.info("----------------------- start CanalClient -----------------------");
	}

	@Override
	public void addAction(String tableName, String primaryKey, Map<String, Object> map) {
		productService.addProductToEs(tableName, primaryKey, map);
	}

	@Override
	public void editAction(String tableName, String primaryKey, Map<String, Object> map) {
		productService.updateProductToEs(tableName, primaryKey, map);
	}

	@Override
	public void deleteAction(String tableName, Map<String, Object> map) {
		productService.deleteProductToEs(tableName, map);
	}

	/**
	 * <p>Title: 读线程</p>
	 * <p>Description: 该线程用于读取客户端数据</p>
	 */

	public class ReaderThread extends Thread {
		@Override
		public void run() {
			while (true) {
				try {
					getData(connector);
				} catch (Exception e) {
					System.out.println(ExceptionUtil.getMessage(e));
					continue;
				}
			}
		}
	}

}
